package com.gukun.springboot.tdengine.service.backuprecover.recover;

import com.gukun.springboot.tdengine.domain.enums.TagType;
import com.gukun.springboot.tdengine.pojo.TagInfo;
import com.gukun.springboot.tdengine.service.backuprecover.entity.Table;
import com.zaxxer.hikari.HikariDataSource;
import lombok.SneakyThrows;
import org.apache.avro.Schema;
import org.apache.avro.SchemaBuilder;
import org.apache.avro.file.DataFileReader;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.hadoop.thirdparty.com.google.common.util.concurrent.RateLimiter;
import org.apache.poi.ss.formula.functions.Rate;

import java.io.File;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author zhang
 */
public class RecoverDbTableTask implements Runnable{
    private final BlockingDeque<File> fileQueue;
    private final String dbName;
    private final AtomicLong tableNum;
    private final HikariDataSource tdEngineConnPool;
    public RecoverDbTableTask(BlockingDeque<File> fileQueue, String dbName, AtomicLong tableNum, HikariDataSource tdEngineConnPool) {
        this.fileQueue = fileQueue;
        this.dbName = dbName;
        this.tableNum = tableNum;
        this.tdEngineConnPool = tdEngineConnPool;
    }

    @SneakyThrows
    @Override
    public void run() {
        try(Connection connection = tdEngineConnPool.getConnection();
            Statement statement = connection.createStatement()){
            statement.execute("USE "+dbName);
            String stableName = null;
            List<GenericRecord> buffer = new ArrayList<>(2500);
            RateLimiter rateLimiter = RateLimiter.create(1500);
            while(true){
                File file = fileQueue.poll(100, TimeUnit.MILLISECONDS);
                if(file == null && fileQueue.isEmpty()){
                    break;
                }
                if(file == null){
                    continue;
                }
                try(DataFileReader<GenericRecord> dataFileReader = new DataFileReader<>(file, new GenericDatumReader<>())){
                    Schema schema = dataFileReader.getSchema();
                    stableName = schema.getName();
                    GenericRecord record = null;
                    while (dataFileReader.hasNext()) {
                        rateLimiter.acquire();
                        record = dataFileReader.next();
                        buffer.add(record);
                        if(buffer.size() >= 2500){
                            writeData(buffer,stableName,statement,tableNum);
                        }
                    }
                }
                if(!buffer.isEmpty()){
                    writeData(buffer,stableName,statement,tableNum);
                }
            }
        }catch (Exception e){
            throw e;
        }
    }

    private void writeData(List<GenericRecord> buffer, String stableName, Statement statement,AtomicLong count) {
        TagInfo tagInfo = null;
        String tn = null;
        String ta = null;
        String desc = null;
        String tableName =null;
        StringBuilder createTableSql = new StringBuilder("CREATE TABLE ");
        GenericRecord record = null;
        for (int i=0;i<buffer.size();i++) {
            record = buffer.get(i);
            tableName = record.get("name").toString();
            desc = record.get("desc") == null ? "" : record.get("desc").toString();
            tn = record.get("tn") == null ? "" : record.get("tn").toString();
            ta = record.get("ta") == null ? null : "\"" + record.get("ta").toString() + "\"";
            tagInfo = new TagInfo();
            tagInfo.setTagName(tableName);
            tagInfo.setTagType(getTagType(stableName));
            tagInfo.setTagDesc(desc);
            createTableSql.append("IF NOT EXISTS ").append(tableName).append(" USING ").append(stableName)
                    .append(" TAGS(\"").append(tagInfo).append("\",\"").append(tn).append("\",").append(ta).append(")");
            if(i<buffer.size()-1){
                createTableSql.append(" ");
            }
        }
        createTableSql.append(";");
        try {
            statement.execute(createTableSql.toString());
            count.addAndGet(buffer.size());
            buffer.clear();
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }

    }

    public TagType getTagType(String stableName){
        return switch (stableName) {
            case "tfloat" -> TagType.FLOAT;
            case "tdouble" -> TagType.DOUBLE;
            case "tint" -> TagType.INT;
            case "tbool" -> TagType.BOOL;
            default -> TagType.FLOAT;
        };
    }
}
